 1.Druid入门案例之从Kafka中加载流式数据
   
   1).数据及需求说明
   Druid典型应用架构
   不在Druid中处理复杂的数据转换清洗工作
   假设有以下网络流量数据：
       ts ：时间戳
	   srcip ：发送端 IP 地址
	   srcport ：发送端端口号
	   dstip ：接收端 IP 地址
	   dstport ：接收端端口号
	   protocol ：协议
	   packets ：传输包的数量
	   bytes ：传输字节数
	   cost ：传输耗费的时间
   数据为json格式，通过Kafka传输
   每行数据包含：时间戳（ts）、维度列、指标列
   维度列：srcip、srcport、dstip、dstport、protocol
   指标列：packets、bytes、cost
   
   需要计算的指标：
	   记录的条数：count
	   packets：max
	   bytes：min
	   cost：sum
   数据汇总的粒度：分钟
   测试数据：
{"ts":"2020-10-01T00:01:35Z","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":1, "bytes":1000, "cost": 0.1}
{"ts":"2020-10-01T00:01:36Z","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":2, "bytes":2000, "cost": 0.1}
{"ts":"2020-10-01T00:01:37Z","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":3, "bytes":3000, "cost": 0.1}
{"ts":"2020-10-01T00:01:38Z","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":4, "bytes":4000, "cost": 0.1}

{"ts":"2020-10-01T00:02:08Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":5, "bytes":5000, "cost": 0.2}
{"ts":"2020-10-01T00:02:09Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":6, "bytes":6000, "cost": 0.2}
{"ts":"2020-10-01T00:02:10Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":7, "bytes":7000, "cost": 0.2}
{"ts":"2020-10-01T00:02:11Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":8, "bytes":8000, "cost": 0.2}
{"ts":"2020-10-01T00:02:12Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":9, "bytes":9000, "cost": 0.2}
   最后执行查询：
select * from tab;

-- 有两个返回值，以下仅为示意
{"ts":"2020-10-01T00:01","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":5, "bytes":1000, "cost": 0.4, "count":4}
{"ts":"2020-10-01T00:02","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":9, "bytes":5000, "cost": 1.0, "count":5}

-- 其他查询
select dstPort, min(packets), max(bytes), sum(count), min(count)
from tab
group by dstPort   
   2).创建Topic发送消息
   启动Kafka集群，并创建一个名为 "lagoudruid1" 的 Topic ：
   -- 创建topic
kafka-topics.sh --create --zookeeper linux121:2181,linux122:2181/myKafka 
--replication-factor 2 --partitions 6 --topic lagoudruid1
   -- 启动生产者
kafka-console-producer.sh --broker-list linux121:9092,linux122:9092 --topic lagoudruid1
   3).摄取数据
   浏览器访问 linux123:8888,点击控制台中的 Load data
   (1).Start
   选择 Apache Kafka ，点击 Connect data
   (2).Connect
   在 Bootstrap servers 输入 linux121:9092,linux122:9092
   在 Topic 输入 lagoudruid1
   点击 Preview 确保看到的数据是正确的
   后点击"Next: Parse data"进入下一步
   (3).Parse data
   数据加载器将尝试自动为数据确定正确的解析器。可以使用多种解析器解析数据
   这里使用 json 解析器解析数据
   (4).Parse time
   定义数据的主时间列
   (5).Tranform
   不建议在Druid中进行复杂的数据转换操作，可考虑将这些操作放在数据预处理
   这里没有定义数据转换
   (6).Filter
   不建议在Druid中进行复杂的数据过滤操作，可考虑将这些操作放在数据预处理
   这里没有定义数据过滤
   (7).configure Schema
   定义指标列、维度列
   定义如何在维度列上进行计算
   定义是否在摄取数据时进行数据的合并（即Rollup），以及Rollup的粒度
   (8).Partition
   定义如何进行数据分区
   Primary partitioning有两种方式
       uniform，以一个固定的时间间隔聚合数据，建议使用这种方式。这里将每天的数据作为一个分区
	   arbitrary，尽量保证每个segments大小一致，时间间隔不固定
   Secondary partitioning
	   Max rows per segment，每个Segment最大的数据条数
	   Max total rows，Segment等待发布的最大数据条数	   
   (9).Tune
   定义任务执行和优化的相关参数
   (10).Publish
   定义Datasource的名称
   定义数据解析失败后采取的动作
   (11).Edit spec
   json串为数据摄取规范。可返回之前的步骤中进行更改，也可以直接编辑规范，并在前面的步骤中看到它
   摄取规范定义完成后，单击 Submit 将创建一个数据摄取任务
   4).数据查询
   数据摄取规范发布后创建 Supervisor
   Supervisor 会启动一个Task，从Kafka中摄取数据
   
   等待一小段时间，Datasource被创建，此时可以进行数据的查询
   
   -- 查看全部的数据
select *
from "lagoutab1"

-- 其他查询
select dstPort, min(sum_packets), max(min_bytes), sum("count") as sumcount, min("count")
from "lagoutab1"
group by dstPort

-- count字段加引号，表示是一个列名(本质是进行转义，否则认为count是一个函数，将报错)
   
   备注：维度相同的数据进行了Rollup
   5).数据摄取规范
{
	"type":"kafka",
	"spec":{
		"ioConfig":Object{...},
		"tuningConfig":Object{...},
		"dataSchema":Object{...}
	}
}
   dataSchema。指定传入数据的Schema
   ioConfig。指定数据的来源和去向
   tuningConfig。指定各种摄取参数   
   
   dataSchema的定义：
   Druid摄入数据规范的核心是dataSchema,dataSchema定义了如何解析输入的数据，并将数据存储到Druid中。
		"dataSchema":{
			"dataSource":"lagoutab1",
			"granularitySpec":{
				"type":"uniform",
				"queryGranularity":"MINUTE",
				"segmentGranularity":"DAY",
				"rollup":true
			},
			"timestampSpec":{
				"column":"ts",
				"format":"iso"
			},
			"dimensionsSpec":Object{...},
			"metricsSpec":Array[6]
		}
   
   dataSource。摄取数据后生成 dataSource 的名称（dataSource是在查询中使用的表）
   granularitySpec。如何创建段和汇总数据
   timestampSpec。设置时间戳的列和格式
   dimensionsSpec。指定数据的维度列
   metricsSpec。指定数据的指标列，以及Rollup时指标列如何计算
   transformSpec。指定数据的转换规则和过滤规则，这里没有定义

   备注：如果没有定义Rollup，在摄取数据时维度和度量之间没有区别
   ioConfig的定义：
   输入数据的数据源在ioConfig中指定,每个任务类型都有自己的ioConfig,这里从 kafka 获取数据,配置如下：
		"ioConfig":{
			"type":"kafka",
			"consumerProperties":{
				"bootstrap.servers":"linux121:9092,linux122:9092"
			},
			"topic":"lagoudruid1",
			"inputFormat":{
				"type":"json"
			},
			"useEarliestOffset":true,
			"appendToExisting":true
		}
		
   tuningConfig的定义
   tuningConfig规范根据摄取任务类型而有所不同。
		"tuningConfig":{
			"type":"kafka"
		}
		
 2.从HDFS中加载数据
   
   hdfs dfs -cat /data/druidlog.dat
-- druidlog.dat
{"ts":"2020-10-01T00:01:35Z","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":1, "bytes":1000, "cost": 0.1}
{"ts":"2020-10-01T00:01:36Z","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":2, "bytes":2000, "cost": 0.1}
{"ts":"2020-10-01T00:01:37Z","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":3, "bytes":3000, "cost": 0.1}
{"ts":"2020-10-01T00:01:38Z","srcip":"6.6.6.6", "dstip":"8.8.8.8", "srcport":6666,
"dstPort":8888, "protocol": "tcp", "packets":4, "bytes":4000, "cost": 0.1}
{"ts":"2020-10-01T00:02:08Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":5, "bytes":5000, "cost": 0.2}
{"ts":"2020-10-01T00:02:09Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":6, "bytes":6000, "cost": 0.2}
{"ts":"2020-10-01T00:02:10Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":7, "bytes":7000, "cost": 0.2}
{"ts":"2020-10-01T00:02:11Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":8, "bytes":8000, "cost": 0.2}
{"ts":"2020-10-01T00:02:12Z","srcip":"1.1.1.1", "dstip":"2.2.2.2", "srcport":6666,
"dstPort":8888, "protocol": "udp", "packets":9, "bytes":9000, "cost": 0.2}

   定义数据摄取规范
       HDFS文件，数据格式 json，时间戳 ts
	   不定义Rollup；保留所有的明细数据
	   Segment granularity：Day
	   DataSource Name：lagoutab3
	   
   数据查询
select * from "lagoutab3"
select protocol, count(*) rowcoun, sum(bytes) as bytes, sum(packets) as packets, max(cost)
as maxcost
from "lagoutab3"
group by protocol